package com.atguigu.tingshu.common.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service

public class KafkaService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送kafka消息
     * @param topic 话题名称
     * @param key 消息Key
     * @param data StringSerializer
     */
    public void sendKafkaMessage(String topic,String key,String data){
        // 1.发送消息后得到异步任务
        CompletableFuture completableFuture = kafkaTemplate.send(topic, key, data);
        // 2.获取异步任务正常回调
        // 3.获取异步任务异常回调
        completableFuture.whenCompleteAsync((t,e) ->{
            if (e != null){
                logger.error("[kafka生产者]发送消息异常:{}",e);
            }else {
                logger.info("[kafka生产者]发送消息成功，话题名称：{}，key:{},消息:{}",topic,key,data);
            }
        });
    }


    /**
     * 发送kafka消息
     * @param topic 话题名称
     * @param data StringSerializer
     */
    public void sendKafkaMessage(String topic,String data){
        // 1.发送消息后得到异步任务
        CompletableFuture completableFuture = kafkaTemplate.send(topic, null, data);
        // 2.获取异步任务正常回调
        // 3.获取异步任务异常回调
        completableFuture.whenCompleteAsync((t,e) ->{
            if (e != null){
                logger.error("[kafka生产者]发送消息异常:{}",e);
            }else {
                logger.info("[kafka生产者]发送消息成功，话题名称：{}，key:{},消息:{}",topic,null,data);
            }
        });
    }

}
